package cn.monkey.state.scheduler.reactive;

import cn.monkey.state.core.reactive.ReactiveStateGroupPool;
import cn.monkey.state.scheduler.AbstractSchedulerManager;
import cn.monkey.state.scheduler.EventPublishSchedulerFactory;
import cn.monkey.state.scheduler.SchedulerManagerConfig;
import cn.monkey.state.scheduler.StateGroupSchedulerFactory;
import reactor.core.scheduler.Scheduler;

public class SimpleReactiveSchedulerManager extends AbstractSchedulerManager {

    protected final Scheduler scheduler;
    protected final ReactiveStateGroupPool stateGroupPool;

    public SimpleReactiveSchedulerManager(Scheduler scheduler,
                                          ReactiveStateGroupPool stateGroupPool,
                                          StateGroupSchedulerFactory stateGroupSchedulerFactory,
                                          EventPublishSchedulerFactory eventPublishSchedulerFactory,
                                          SchedulerManagerConfig managerConfig) {
        super(stateGroupSchedulerFactory, eventPublishSchedulerFactory, managerConfig);
        this.stateGroupPool = stateGroupPool;
        this.scheduler = scheduler;
    }

    protected void findBestSchedulerAndTryAddStateGroup(String groupId, Object event, Object... args) {
        this.stateGroupPool.findOrCreate(groupId, args)
                .subscribeOn(this.scheduler)
                .subscribe(fetchStateGroup -> super.findEventPublisherScheduler(groupId)
                        .publish(() -> super.onFetchStateGroup(fetchStateGroup, event)));
    }

    @Override
    public void addEvent(String groupId, Object event, Object... args) {
        if (!this.isRunning.get()) {
            throw new UnsupportedOperationException("schedulerManager has stopped");
        }
        this.findBestSchedulerAndTryAddStateGroup(groupId, event, args);
    }
}
